package com.dongdongshop.mq;

import com.alibaba.dubbo.config.annotation.Reference;
import com.dongdongshop.service.FreemarkerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
import org.springframework.stereotype.Component;

import java.lang.annotation.Annotation;
import java.util.List;

@Component
@RocketMQMessageListener(topic = "id",consumerGroup = "myconsumer-group")
public class RokcetMQconsumer implements RocketMQListener<Long[]>, RocketMQPushConsumerLifecycleListener {
    @Reference
    private FreemarkerService freemarkerService;
    @Override
    public void prepareStart(DefaultMQPushConsumer defaultMQPushConsumer) {
        defaultMQPushConsumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                try {
                    MessageExt messageExt = list.get(0);
                    byte[] bodys = messageExt.getBody();
                    String s = new String(bodys);

                    s = s.substring(1);
                    int length = s.length();
                    s = s.substring(0, length-1);
                    System.out.println(s);
                    String[] goodsids = s.split(",");
                    freemarkerService.createPage(goodsids);

                } catch (Exception e) {
                   return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    @Override
    public void onMessage(Long[] longs) {

    }
}
